package com.dduggs.rocketmq;

import com.alibaba.rocketmq.client.exception.*;
import com.alibaba.rocketmq.client.producer.*;
import com.alibaba.rocketmq.common.message.*;
import com.dduggs.utils.HessianUtils;
import com.dduggs.utils.PropertyFileUtil;
import org.apache.commons.lang3.*;
import org.slf4j.*;
import org.springframework.stereotype.*;

import javax.annotation.*;
import java.io.*;
import java.net.*;

@Service(value = "msgSender")
public class MsgSenderService implements MsgSender{

	private static final Logger LOGGER = LoggerFactory.getLogger(MsgSenderService.class);

	private DefaultMQProducer producer;

	/**
	 * 测试模式，实际不发送
	 */
	private boolean test = false;

	@PostConstruct
	public void init(){

		String nameServer = PropertyFileUtil.get("rocketmq.namesrv.domain");

		//为本地测试关闭metaq
		if (StringUtils.isBlank(nameServer) || "localTest".equals(nameServer)){
			test = true;
			return;
		}

		producer = new DefaultMQProducer("isesol");

		//设置nameserver 地址
		if (StringUtils.isBlank(System.getProperty("rocketmq.namesrv.domain"))){
			System.setProperty("rocketmq.namesrv.domain", nameServer);
		}

		try{

			producer.setNamesrvAddr(nameServer);
			producer.setInstanceName("DEFAULT_MSG_SENDER-" + InetAddress.getLocalHost().getHostName());
			producer.setVipChannelEnabled(false);
			producer.setSendMessageWithVIPChannel(false);
			producer.start();
			LOGGER.info("MQ 发送端启动成功,rocketmq.namesrv.domain={}",
			            System.getProperty("rocketmq.namesrv.domain"));

		} catch (UnknownHostException e){

			LOGGER.error("getHostName error", e);

		} catch (MQClientException e){

			LOGGER.error("MQ 发送端启动失败!nameServer={},excepton={} "
					,System.getProperty("rocketmq.namesrv.domain"), e.getMessage(), e);
		}
	}

	@Override
	public SendResult sendMessage(Serializable message, Class<? extends Topic> topicType){

		Topic topic = RocketMqUtils.getTopic(topicType);

		return sendMessage(message, topic.getTopic(), topic.getTags());
	}

	/**
	 * 发送消息服务
	 *
	 * @param message
	 * 		消息对象
	 * @param topic
	 * 		topic名称
	 * @param tag
	 * 		tag名称
	 *
	 * @return 发送结果
	 */
	@Override
	public SendResult sendMessage(Serializable message, String topic, String tag){

		if (test){

			SendResult sendResult = new SendResult();
			sendResult.setSendStatus(SendStatus.SEND_OK);

			return sendResult;
		}

		try{

			LOGGER.debug("发送MQ消息, message={}", message.toString());

			Message    msg        = new Message(topic, tag, HessianUtils.encode(message));
			SendResult sendResult = producer.send(msg);

			LOGGER.info("sendResult={},topic={},tags={},msgId={}", sendResult.getSendStatus(), topic,tag,sendResult.getMsgId());

			if(sendResult.getSendStatus()!=SendStatus.SEND_OK){

				LOGGER.warn("MQ send fail,topic={},tags={},msgId=", sendResult.getSendStatus(), topic,tag,sendResult.getMsgId());
			}

			return sendResult;

		} catch (IOException ioe){

			LOGGER.error("MQ 发送消息失败 序列化失败   message={} ", message, ioe);

			throw new RuntimeException("MQ 发送消息失败 序列化失败   message=" + message);

		} catch (Exception e){

			LOGGER.error("MQ 发送消息失败  message={} ", message,e);

			throw new RuntimeException(e.getMessage(), e);
		}
	}

	@PreDestroy
	public void destroy(){

		if (producer != null){

			producer.shutdown();

			LOGGER.info("MQ 发送端关闭成功;rocketmq.namesrv.domain={}",
			            System.getProperty("rocketmq.namesrv.domain"));
		}
	}
}